package org.apache.hadoop.hive.cassandra.output.cql;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.hadoop.hive.cassandra.serde.AbstractCassandraSerDe;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datastax.driver.core.BatchStatement;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;

public class DatastaxCqlMapMapper extends AbstractCqlMapper{

	public static final Logger LOGGER = LoggerFactory.getLogger(DatastaxCqlMapMapper.class);
	
	private Session session; 
	
	private Map<String, String> column_types_map = new HashMap<String, String>();
	
	private List<String> needInsertColumns = new ArrayList<String>();
	
	protected final static List<MapWritable> dataMap = new ArrayList<MapWritable>();
	
	ConcurrentMap<String, PreparedStatement> preparedQueries = new ConcurrentHashMap<String, PreparedStatement>();
	
	private Set<Long> set = new HashSet<Long>();
	
	PreparedStatement ps = null;
	
	private List<BoundStatement> BOUNDSTATEMENTCACHE = new ArrayList<BoundStatement>();
	
	
	public DatastaxCqlMapMapper(Session session, JobConf jobConf) {
		super(jobConf);
		this.session = session;
		
		LOGGER.info("列名称:"+jobConf.get(AbstractCassandraSerDe.CASSANDRA_COL_MAPPING));
		LOGGER.info("列类型:"+jobConf.get(AbstractCassandraSerDe.CASSANDRA_COL_TYPE_MAPPING));
		
		String columnsStr = jobConf.get(AbstractCassandraSerDe.CASSANDRA_COL_MAPPING);
        String columnsTypes = jobConf.get(AbstractCassandraSerDe.CASSANDRA_COL_TYPE_MAPPING);
        String[] columnSysTypes = columnsTypes.split(":");
        String[] columnSysNames = columnsStr.split(",");
        if (columnSysTypes.length != columnSysNames.length) {
        	throw new RuntimeException("获取到的hive表列名不能比列类型多");
        }
        int len = columnSysNames.length;
        for(int i=0;i<len;i++){
        	column_types_map.put(columnSysNames[i], columnSysTypes[i]);
        	needInsertColumns.add(columnSysNames[i]);
        }
        StringBuffer sql = new StringBuffer();
		StringBuilder valuesBuilder = new StringBuilder(" VALUES (");
		sql.append("insert into ");
		sql.append(jobConf.get(AbstractCassandraSerDe.CASSANDRA_CF_NAME));
		sql.append("(");
		Iterator<String> iter0 = column_types_map.keySet().iterator();
        while (iter0.hasNext()) {
            String columnName = iter0.next();
            sql.append(columnName);
            valuesBuilder.append("?");
            if (iter0.hasNext()) {
                sql.append(",");
                valuesBuilder.append(",");
            }
        }
        sql.append(")");
        valuesBuilder.append(")");
        sql.append(valuesBuilder);
        LOGGER.info("datax batch insert sql is :{}",sql.toString());
        ps = session.prepare(sql.toString());
	}
	

	@Override
	public void writeToCassandra(Writable writable) throws IOException {
		MapWritable hiveData = (MapWritable)writable;//不能直接cache 这个引用导致list所有对象是一个
		BoundStatement bs = ps.bind();
    	Iterator<String> keys = needInsertColumns.iterator();
    	//Writable textWritableValue = null;
    	while(keys.hasNext()){
    		String column = keys.next();//列名称
    		Text columnName = new Text(column);
    		//LOGGER.info("列名:{}" , columnName);
    		if(hiveData.containsKey(columnName)){
    			Writable textWritableValue = hiveData.get(columnName);
    			String columnType = column_types_map.get(column);
    			//LOGGER.info("列类型为:{}" , columnType);
    			if(org.apache.hadoop.hive.serde.Constants.STRING_TYPE_NAME.equals(columnType)
    				   ||org.apache.hadoop.hive.serde.Constants.CHAR_TYPE_NAME.equals(columnType)	
    				   ||org.apache.hadoop.hive.serde.Constants.VARCHAR_TYPE_NAME.equals(columnType)
    					){
    				bs.setString(column, textWritableValue.toString());
    			}else if(org.apache.hadoop.hive.serde.Constants.BIGINT_TYPE_NAME.equals(columnType)){
    				bs.setLong(column, Long.parseLong(textWritableValue.toString()));
    				set.add(Long.parseLong(textWritableValue.toString()));
    			}else if(org.apache.hadoop.hive.serde.Constants.INT_TYPE_NAME.equals(columnType)
    					||org.apache.hadoop.hive.serde.Constants.TINYINT_TYPE_NAME.equals(columnType)
    					||org.apache.hadoop.hive.serde.Constants.SMALLINT_TYPE_NAME.equals(columnType)
    					){
    				bs.setInt(column, Integer.parseInt(textWritableValue.toString()));
    			}else if(org.apache.hadoop.hive.serde.Constants.DOUBLE_TYPE_NAME.equals(columnType)){
    				bs.setDouble(column, Double.parseDouble(textWritableValue.toString()));
    			}else if(org.apache.hadoop.hive.serde.Constants.FLOAT_TYPE_NAME.equals(columnType)){
    				bs.setFloat(column, Float.parseFloat(textWritableValue.toString()));
    			}else if(org.apache.hadoop.hive.serde.Constants.BOOLEAN_TYPE_NAME.equals(columnType)){
    				bs.setBool(column, Boolean.parseBoolean(textWritableValue.toString()));
    			}else if(org.apache.hadoop.hive.serde.Constants.DECIMAL_TYPE_NAME.equals(columnType)){
    				bs.setDecimal(column,new BigDecimal(textWritableValue.toString()));
    			}else if(org.apache.hadoop.hive.serde.Constants.DATE_TYPE_NAME.equals(columnType)){
    				bs.setTimestamp(column, new Date(textWritableValue.toString()));
    			}else {
    				LOGGER.error("找不到的列类型:{}={}" , column , columnType);
    			}
    		}else {
    			LOGGER.error("不存在的列:{}" ,columnName);
    		}
    	}
    	BOUNDSTATEMENTCACHE.add(bs);
		int size = BOUNDSTATEMENTCACHE.size();
		if(batchInsertSize == size){
			this.write2cassandra();
			BOUNDSTATEMENTCACHE.clear();
			allData.addAndGet(size);
			//LOGGER.info(""+Thread.currentThread().getName() + " batch insert 2 casaandra " +allData.get()  + ":data"  );
		}
	}


	@Override
	public void finishWriteJob() throws IOException {
		if(!BOUNDSTATEMENTCACHE.isEmpty()){
			int size = BOUNDSTATEMENTCACHE.size();
			this.write2cassandra();
			BOUNDSTATEMENTCACHE.clear();
			allData.addAndGet(size);
			LOGGER.info(""+Thread.currentThread().getName() + "flush "+size+":data and all batch insert 2 casaandra " +allData.get()  + ":data"  );
		}
	}
	
	


	@Override
	public void write2cassandra(){
		if(!BOUNDSTATEMENTCACHE.isEmpty()){
	        BatchStatement batch = new BatchStatement();
	        for(BoundStatement bs : BOUNDSTATEMENTCACHE){
	        	batch.add(bs);
	        }
	        ResultSet rs = session.execute(batch);
	        //LOGGER.info("batch insert result:{},key数目{}",rs.toString(),set.size());
		}
	}
}
